#include "raft/peer.h"

#include <assert.h>
#include <google/protobuf/text_format.h>

#include <algorithm>
#include <climits>
#include <string>
#include <vector>

#include "raft/apply.h"
#include "raft/context.h"
#include "raft/meta.h"
#include "raft/primary.h"
#include "raft/raft.h"
#include "raft/raft.pb.h"
#include "raft/raft_log.h"
#include "zrpc/base/logger.h"
#include "zrpc/net/socket.h"

namespace libraft {
Peer::Peer(Raft* raft, const std::string& server_addr, const std::string& ip,
           const int16_t port, zrpc::EventLoop* loop,
           const std::shared_ptr<Context>& context,
           const std::shared_ptr<Meta>& meta,
           const std::shared_ptr<RaftLog>& raft_log, const Options& options)
    : raft_(raft),
      server_addr_(server_addr),
      ip_(ip),
      port_(port),
      loop_(loop),
      context_(context),
      meta_(meta),
      raft_log_(raft_log),
      options_(options),
      next_index_(1),
      match_index_(0),
      client_last_op_time_(0),
      incr_(0) {
  next_index_ = raft_log_->LastLogIndex() + 1;
  match_index_ = context_->last_applied.load();

  tcp_client_.reset(new zrpc::TcpClient(loop, ip, port, nullptr)),
      rpc_channel_.reset(new zrpc::RpcChannel());
  cmd_service_stub_.reset(new raft::CmdMessageService_Stub(rpc_channel_.get()));

  tcp_client_->SetConnectionCallback(
      std::bind(&Peer::OnConnection, this, std::placeholders::_1));

  tcp_client_->SetMessageCallback(
      std::bind(&zrpc::RpcChannel::OnMessage, rpc_channel_.get(),
                std::placeholders::_1, std::placeholders::_2));

  tcp_client_->EnableRetry();
  tcp_client_->Connect(true);
}

Peer::~Peer() {}

void Peer::OnConnection(const std::shared_ptr<zrpc::TcpConnection>& conn) {
  if (conn->Connected()) {
    rpc_channel_->SetConnection(conn);
  } else {
  }
}

void Peer::DelayAppendEntriesTask() {
  loop_->DelayRunInLoop(std::bind(&Peer::AppendEntriesRPC, this));
}

void Peer::AddAppendEntriesTask() {
  loop_->RunInLoop(std::bind(&Peer::AppendEntriesRPC, this));
}

void Peer::AddRequestVoteTask() {
  loop_->RunInLoop(std::bind(&Peer::RequestVoteRPC, this));
}

void Peer::ResponseAppendEntriesRPC(raft::CmdResponse* resp) {
  std::unique_lock<std::mutex> lk(context_->global_mutex_);
  if (resp->append_entries_res().term() > context_->current_term) {
    context_->BecomeFollower(resp->append_entries_res().term());
    context_->voted_for_ip.clear();
    context_->voted_for_port = 0;
    meta_->SetCurrentTerm(context_->current_term);
    meta_->SetVotedForIp(context_->voted_for_ip);
    meta_->SetVotedForPort(context_->voted_for_port);
  } else if (resp->append_entries_res().term() < context_->current_term) {
    // ignore old term msg
    // 让对方触发领导选举，更新term
    return;
  }

  if (context_->role == Role::kLeader) {
    if (resp->append_entries_res().success()) {
      if (resp->append_entries_res().num_entries() > 0) {
        match_index_ = resp->append_entries_res().prev_log_index() +
                       resp->append_entries_res().num_entries();
        if (resp->append_entries_res().last_log_term() ==
            context_->current_term) {
          if (raft_->AdvanceLeaderCommitIndex()) {
            raft_->ScheduleApply();
          }
        }

        next_index_ = resp->append_entries_res().prev_log_index() +
                      resp->append_entries_res().num_entries() + 1;
        LOG_DEBUG << "resp append ok:"
                  << resp->append_entries_res().last_log_term() << " "
                  << context_->current_term << " " << match_index_ << " "
                  << next_index_ << " " << options_.local_ip << " "
                  << options_.local_port;
      }
    } else {
      uint64_t adjust_index = std::min(
          resp->append_entries_res().last_log_index() + 1, next_index_ - 1);
      if (adjust_index > 0) {
        next_index_ = adjust_index;
        DelayAppendEntriesTask();
      }

      if (next_index_ < raft_log_->FirstLogIndex()) {
      }
    }
  } else if (context_->role == Role::kFollower) {
  } else if (context_->role == Role::kCandidate) {
  }
}

void Peer::AppendEntriesRPC() {
  uint64_t prev_log_index = 0;
  uint64_t num_entries = 0;
  uint64_t prev_log_term = 0;
  uint64_t last_log_index = 0;
  uint64_t current_term = 0;

  raft::CmdRequest req;
  raft::CmdRequest_AppendEntries* append_entries = req.mutable_append_entries();

  {
    std::unique_lock<std::mutex> lk(context_->global_mutex_);
    prev_log_index = next_index_ - 1;
    last_log_index = raft_log_->LastLogIndex();
    /*if (next_index_ > last_log_index &&
            client_last_op_time_ + options_.heart_beat_us_ >
                    zrpc::TimeStamp::NowMicros()) {
      return;
    }*/

    client_last_op_time_ = zrpc::TimeStamp::NowMicros();

    if (prev_log_index != 0) {
      raft::Entry entry;
      assert(raft_log_->GetEntry(prev_log_index, &entry));
      prev_log_term = entry.term();
    }

    current_term = context_->current_term;

    req.set_type(raft::Type::kAppendEntries);
    append_entries->set_ip(options_.local_ip);
    append_entries->set_port(options_.local_port);
    append_entries->set_term(current_term);
    append_entries->set_prev_log_index(prev_log_index);
    append_entries->set_prev_log_term(prev_log_term);
    append_entries->set_leader_commit(context_->commit_index);
  }

  for (uint64_t index = next_index_; index <= last_log_index; index++) {
    raft::Entry tmp_entry;
    assert(raft_log_->GetEntry(index, &tmp_entry));
    raft::Entry* entry = append_entries->add_entries();
    *entry = tmp_entry;
    num_entries++;
  }

  raft::CmdResponse* res = new raft::CmdResponse;
  cmd_service_stub_->CmdMessage(
      nullptr, &req, res,
      NewCallback(this, &Peer::ResponseAppendEntriesRPC, res));
}

void Peer::ResponseVoteRPC(raft::CmdResponse* res) {
  std::unique_lock<std::mutex> lk(context_->global_mutex_);
  const raft::CmdResponse_RequestVoteResponse& res_vote =
      res->request_vote_res();
  if (res_vote.term() > context_->current_term) {
    LOG_INFO << "become follower:" << options_.local_ip << " "
             << options_.local_port;
    context_->BecomeFollower(res_vote.term());
    context_->voted_for_ip.clear();
    context_->voted_for_port = 0;
    meta_->SetCurrentTerm(context_->current_term);
    meta_->SetVotedForIp(context_->voted_for_ip);
    meta_->SetVotedForPort(context_->voted_for_port);
    return;
  }

  if (context_->role == Role::kCandidate) {
    if (res_vote.vote_granted()) {
      if (CheckAndVote(res_vote.term())) {
        LOG_INFO << "become leader:" << options_.local_ip << " "
                 << options_.local_port;
        context_->BecomeLeader();
        raft_->UpdatePeerInfo();
        raft_->AddPrimaryTask(kHeartBeat, false);
      }
    } else {
      LOG_INFO << "become follower:" << options_.local_ip << " "
               << options_.local_port;
      context_->BecomeFollower(res_vote.term());
      context_->voted_for_ip.clear();
      context_->voted_for_port = 0;
      meta_->SetCurrentTerm(context_->current_term);
      meta_->SetVotedForIp(context_->voted_for_ip);
      meta_->SetVotedForPort(context_->voted_for_port);
    }
  } else if (context_->role == Role::kFollower) {
  } else if (context_->role == Role::kLeader) {
  }
}

void Peer::RequestVoteRPC() {
  uint64_t last_log_term;
  uint64_t last_log_index;

  raft::CmdRequest req;
  {
    std::unique_lock<std::mutex> lk(context_->global_mutex_);
    raft_log_->GetLastLogTermAndIndex(&last_log_term, &last_log_index);

    req.set_type(raft::Type::kRequestVote);
    raft::CmdRequest_RequestVote* request_vote = req.mutable_request_vote();
    request_vote->set_ip(options_.local_ip);
    request_vote->set_port(options_.local_port);
    request_vote->set_term(context_->current_term);
    request_vote->set_last_log_term(last_log_term);
    request_vote->set_last_log_index(last_log_index);
  }

  raft::CmdResponse* res = new raft::CmdResponse;
  cmd_service_stub_->CmdMessage(nullptr, &req, res,
                                NewCallback(this, &Peer::ResponseVoteRPC, res));
}

bool Peer::CheckAndVote(uint64_t vote_term) {
  if (context_->current_term != vote_term) {
    return false;
  }
  return (++context_->vote_quorum) > (options_.members.size() / 2);
}
}  // namespace libraft